package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.UserLoginBean;
import com.atguigu.gmall.realtime.common.bean.UserRegisterBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 从 Kafka 页面日志主题读取数据，统计七日回流用户和当日独立用户数。
 * 之前的活跃用户，一段时间未活跃（流失），今日又活跃了，就称为回流用户。
 * 此处要求统计回流用户总数。规定当日登录，且自上次登录之后至少7日未登录的用户为回流用户。
 */
public class DwsUserUserLoginWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserLoginWindow().start(10024, 4, "dws_user_user_login_window", Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 读取 Kafka 页面主题数据
        // 转换数据结构 流中数据由 String 转换为 JSONObject。
        SingleOutputStreamOperator<JSONObject> filterDs = stream.map(JSON::parseObject)
                .filter(
                        // 过滤数据
                        // 统计的指标与用户有关,uid不为null的数据才是有用的
                        // 两种情况
                        // 1.用户打开应用后自动登录       -----登录操作发生在会话首页，所以保留首页即可；
                        // 2.用户打开应用后没有登录，浏览部分页面后跳转到登录页面，中途登录     -----登录操作发生在 login 页面，login页面之后必然会跳转到其它页面，保留login之后的页面即可记录情况2的登录操作。
                        // 我们应保留 uid 不为 null 且 last_page_id 为 null 或 last_page_id 为 login 的浏览记录。
                        new FilterFunction<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject jsonObject) throws Exception {
                                // 获取uid
                                String uid = jsonObject.getJSONObject("common").getString("uid");
                                // 获取last_page_id
                                String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");

                                return uid != null && (last_page_id == null || last_page_id.equals("login"));
                            }
                        }
                );
        // filterDs.print("FILTERDS");


        // 设置水位线


        // 按照uid分组
        SingleOutputStreamOperator<UserLoginBean> processDs = filterDs.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObject) throws Exception {
                        return jsonObject.getJSONObject("common").getString("uid");
                    }
                }
        ).process(
                new KeyedProcessFunction<String, JSONObject, UserLoginBean>() {
                    private ValueState<String> lastLonginDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext runtimeContext = getRuntimeContext();
                        ValueStateDescriptor<String> lastLonginDateStateDesc = new ValueStateDescriptor<>("lastLonginDateStateDesc", Types.STRING);
                        lastLonginDateState = runtimeContext.getState(lastLonginDateStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, UserLoginBean>.Context context, Collector<UserLoginBean> collector) throws Exception {
                        Long ts = jsonObject.getLong("ts");
                        String today = DateFormatUtil.tsToDate(ts);
                        String lastLoginDate = lastLonginDateState.value();
                        Long uuCt = 0L;
                        Long backCt = 0L;
                        // 统计回流用户数和独立用户数
                        // 使用状态编程, 记录用户末次登录日期
                        // 如果状态不为null : 1. 末次登录日期不等于null也不等于当前日期,uuCt赋1, 更新状态今日:当天日期与末次日期大于等于八天为回流用户,
                        //                      backCt赋1,反之为0
                        //                  2. 末日登录日期为当天, uuCt和backCt都赋0, 舍弃数据
                        //                  3, 末次日期为null, uuCt为1, backCt为0, 更新状态为今日
                        if (!today.equals(lastLoginDate)) {
                            // 今天第一次登录
                            uuCt = 1L;
                            lastLonginDateState.update(today);
                            if (lastLoginDate != null) {
                                Long lastLoginTs = DateFormatUtil.dateTimeToTs(lastLoginDate);
                                if ((ts - lastLoginTs) / 1000 / 60 / 60 / 24 > 7) {
                                    backCt = 1L;
                                }
                            }
                        }
                        if (uuCt == 1) {
                            UserLoginBean userLoginBean = UserLoginBean.builder()
                                    .stt("")
                                    .edt("")
                                    .curDate("")
                                    .backCt(backCt)
                                    .uuCt(uuCt)
                                    .ts(ts)
                                    .build();
                            collector.collect(userLoginBean);
                        }

                    }
                }
        );
        // processDs.print("processDs");


        // 开窗 聚合
        SingleOutputStreamOperator<UserLoginBean> reduceDs = processDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserLoginBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(120L))
        ).windowAll(
                TumblingEventTimeWindows.of(Time.seconds(5L))
        ).reduce(
                new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean t1, UserLoginBean t2) throws Exception {
                        t1.setBackCt(t1.getBackCt() + t2.getBackCt());
                        t1.setUuCt(t1.getUuCt() + t2.getUuCt());
                        return t1;
                    }
                },
                new ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>.Context context, Iterable<UserLoginBean> iterable, Collector<UserLoginBean> collector) throws Exception {
                        UserLoginBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        collector.collect(bean);
                    }
                }
        );

        // reduceDs.print("REDUCEDS");


        // 写入doris
        reduceDs.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, "dws_user_user_login_window"));



    }
}
